package cn.z2huo.demo.kafka.apache.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.Scanner;

/**
 * <p>异步发送消息的生产者
 *
 * @author z2huo
 */
@Slf4j
public class AsyncProducer {

    public static final String TOPIC = "test-topic";

    public static void main(String[] args) {
        new AsyncProducer().send();
    }

    public void send() {
        Properties props = ProducerProperties.getProducerProperties();

        // 创建 Kafka 生产者
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String line = scanner.nextLine();
                String message = line + " " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
                log.info("message is {}", message);
//                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, 0, "partition-0", line);
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, message);
                producer.send(record, new AsyncSendMessageCallback());
            }
            // 刷新并关闭生产者
            producer.flush();
        }
    }

}
